package com.atguigu.transform;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import com.atguigu.pojo.Wordcount;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/13 22:24
 *
 * 用户自定义函数
 *      MapFunction
 *      FilterFunction
 *      FlatMapFunction
 *      ReduceFunction
 *      ....
 *
 *
 *  自定义函数的方式
 *          1. 自定义类，实现函数接口
 *          2. 匿名内部类的方式
 *          3.lambda
 *
 */
public class Flink04_UserDefineFunction {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);
        //获取数据

        // 读取socket文本数据，将数据封装成Event，打印输出
        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);

        SingleOutputStreamOperator<Event> eventDs = ds.map(
                line -> {
                    String[] fields = line.split(",");
                    return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

                }

        );
        //eventDs.print();

        //匿名内部类方式
        SingleOutputStreamOperator<Event> eventDs1 = ds.map(
                new MapFunction<String, Event>() {
                    @Override
                    public Event map(String line) throws Exception {
                        String[] fields = line.split(",");
                        return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

                    }
                }
        );
        //eventDs1.print();


        // 自定义类
        ds.map(
                new MyMapFunction(",")
        ).print();



        env.execute();




    }
    //自定义类
    public static class MyMapFunction implements MapFunction<String,Event>{

        private String seperator;

        public MyMapFunction(String seperator){
            this.seperator = seperator;
        }

        @Override
        public Event map(String line) throws Exception {
            String[] fields = line.split(seperator);
            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

        }
    }
}
